package com.study;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.TimeoutException;

public class PublishConfirmsProducer {

    public static void main(String[] args) throws NoSuchAlgorithmException, KeyManagementException, URISyntaxException, IOException, TimeoutException, InterruptedException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUri(URI.create("amqp://root:123456@192.168.181.131:5672/%2f"));

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        //声明发送方确认机制
        AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();

        channel.queueDeclare("queue.confirm",false,false,false,null);
        channel.exchangeDeclare("exchange.confirm", "direct",false,false,null);

        channel.queueBind("queue.confirm","exchange.confirm","confirmroutingkey.key");


        String msg  = "hello world";
        //批处理的设置大小
        int batchSize = 10;
        //对用于需要等待消息确认的计数
        int outStrandingconfirms = 0;


        for (int i = 0; i < 103; i++) {
            channel.basicPublish("exchange.confirm","confirmroutingkey.key",null,(msg+i).getBytes());
            outStrandingconfirms++;
            if(outStrandingconfirms == batchSize){
                //此时已经有一个批次的消息需要等待确认
                channel.waitForConfirms(5000);
                System.out.println("消息已经被确认了");
                outStrandingconfirms = 0;

            }
        }


        if(outStrandingconfirms > 0){
            channel.waitForConfirms(5000);
            System.out.println("剩余消息已经被确认了");
        }

        channel.close();
        connection.close();
    }
}
